package com.central.service

import java.sql.{DriverManager, Timestamp}
import java.text.SimpleDateFormat
import java.util
import java.util.{Date, Properties, UUID}

import com.alibaba.fastjson.{JSONArray, JSONObject}
import com.central.bean.WifiEntity
import com.central.utils.{ConfigUtil, LoadESData, ParseTime}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.elasticsearch.spark.rdd.EsSpark

class LingerService extends Serializable {
  val TABLE_LINGER_STATUS = ConfigUtil.properties.getProperty("table.linger.status")
  val TABLE_LINGER_RESULT = ConfigUtil.getProperties().getProperty("table.linger.result")
  val TABLE_LINGER_DETAIL = ConfigUtil.getProperties().getProperty("table.linger.detail")
  val LINGER_ES_INDEX: String = ConfigUtil.getProperties().getProperty("collide.es.index")
  val JDBC_URL = ConfigUtil.getProperties().getProperty("mysql.url")
  val USER_NAME = ConfigUtil.getProperties().getProperty("mysql.username")
  val PASSWORD = ConfigUtil.getProperties().getProperty("mysql.password")

  def lingerAnalysis(spark: SparkSession, condition: LingerCondition): Unit = {
    val properties = new Properties()
    properties.put("user", "root")
    properties.put("password", "root")
    val gang = condition.hasGang.toInt
    val hasCriminalRecord = condition.hasCriminalRecord.toInt
    val jobId = condition.jobId
    val incidentTime = condition.incidentTime
    val incidentDay = condition.incidentDay
    val codeType = condition.codeType
    val lingerTime = condition.lingerTime
    val lingerDay = condition.lingerDay
    var deviceCodes: util.ArrayList[String] = new util.ArrayList()
    condition.deviceCodes.split(",").map(deviceCodes.add(_))
    //通过案发时间、案发地点找出对应数据
    val incidentRdd = loadIncidentData(spark, LINGER_ES_INDEX, incidentTime, incidentDay, deviceCodes, codeType)
    println(incidentRdd.count()+" incident rdd")
    //通过徘徊分析条件，找出需要进行徘徊分析的数据
    val lingerRdd = loadLingerData(spark, LINGER_ES_INDEX, lingerTime, lingerDay, codeType)
        .map(x=>{
          (x.idNumber,x)
        })
    println(lingerRdd.count()+" linger rdd")

    println(lingerRdd.map(x => {
      (x._1, 1)
    }).reduceByKey(_ + _)
      .map(_._1)
      .count()+" linger code")
    //需要进行徘徊分析的数据肯定是在案发条件数据里，所以找出案发时间与徘徊时间数据交集
    val lingerJoinRdd =
      lingerRdd
          .join(incidentRdd)

      .map(x => {
        (x._2._1.idNumber, x._2._1)
      }).cache()
    println(lingerJoinRdd.count()+"join count")
    //再对交集数据进行团伙分析
    val lingerCrimeRdd = if (gang == 1) {
      if (hasCriminalRecord==1) {
        val crimeRdd: RDD[(String, String)] = new GangFindService().loadCrimeData(spark)
        lingerJoinRdd.join(crimeRdd)
          .map(x => {
            (x._2._1.idNumber, x._2._1)
          })
      }
      else lingerJoinRdd

    }
    else lingerJoinRdd
    println(lingerCrimeRdd.count()+" after crime count")

    //完成团伙分析的数据再进行徘徊分析
//    lingerCrimeRdd.groupByKey().map(_._2.size).collect().toList.map(println(_))
    val resultRdd: RDD[(Int, List[(WifiEntity, WifiEntity, WifiEntity)], String)] = lingerCrimeRdd.groupByKey()
      .map(x => {
        x._2.toList
      })
      .filter(x=>{
        x.size>3
      })

      .map(entities => {
        val resultid = UUID.randomUUID() + ""
        val sort = entities.sortBy(_.captime)
        val tuple: (Int, List[(WifiEntity, WifiEntity, WifiEntity)]) = vecCaculator(sort)
        (tuple._1, tuple._2, resultid)
      })
      .filter(x=>{
        !x._2.isEmpty
      })
      .cache()
    //      .map(x => {
    //        val code = x._1
    //        val entities = x._2.toList.sortBy(x => {
    //          x.captime
    //        })
    //
    //      })
    println(resultRdd.count())
    import spark.implicits._
    resultRdd.map(x => {
      val entity = x._2.head._1
      val lingerCount = x._1
      val tagcode = entity.idNumber
      val location = entity.location
      val codetype = entity.idNumberType
      val resultid = x._3
      val createtime = ParseTime.LongToDate(System.currentTimeMillis() + "")
      val phone = entity.phoneNumber
      LingerResult(resultid, codetype, 1, createtime, jobId, lingerCount, location, phone, tagcode)

    }).toDF()
      .write
      .mode(SaveMode.Append)
      .jdbc(JDBC_URL, TABLE_LINGER_RESULT, properties)


    val detailRdd: RDD[LingerDetail] = resultRdd.flatMap(x => {
      val entites = x._2
      val resultid = x._3
      val wifientity: List[WifiEntity] = entites.flatMap(x => {
        x.productIterator.toList.map(ele => {
          ele.asInstanceOf[WifiEntity]
        })
      }).distinct

      wifientity.map(wifi => {
        LingerDetail(wifi.address, wifi.captime, wifi.idNumberType, wifi.deviceNumber.toInt, wifi.deviceName, wifi.latitude, wifi.location, wifi.longitude, wifi.phoneNumber, resultid, wifi.idNumber)
      })
    })
//    println(detailRdd.count())
    detailRdd
      .toDF()
      .write
      .mode(SaveMode.Append)
      .jdbc(JDBC_URL, TABLE_LINGER_DETAIL, properties)

    updateStatus(jobId)
  }

  /**
    * 徘徊分析
    *
    * @param data
    * @return
    */
  def vecCaculator(data: List[WifiEntity]) = {
    var count = 0
    val afterRdd = data
      .zipWithIndex
      .filter(_._2 > 1)
      .map(x => {
        x._1
      })
    val zipRdd: List[(WifiEntity, WifiEntity)] = data.zip(afterRdd)
    val vecRdd = zipRdd
      .map(x => {
        //        x._1表示前一个点,
        ((x._1.longitude, x._1.latitude), (x._2.longitude, x._2.latitude), x)
      })
      //将拉链后的结果数据，转换成向量。即相邻两个点经纬度坐标换算成一个向量。
      .map(x => {

      (x._2._1 - x._1._1, x._2._2 - x._1._2, x._3)
    })
    //再将数据进行拉链，得到相邻向量组成的结果
    val aftervecRdd = vecRdd.zipWithIndex.filter(_._2 > 1).map(_._1)
    val zipVecRdd = vecRdd.zip(aftervecRdd)
    val result: List[(Double, (WifiEntity, WifiEntity, WifiEntity))] = zipVecRdd.map(x => {
      //      val pre: (WifiEntity, WifiEntity) = x._1._3
      //      val after: (WifiEntity, WifiEntity) = x._2._3
      val threePoint = (x._1._3._1, x._1._3._2, x._2._3._2)
      val x1 = x._1._1
      val y1 = x._1._2
      val x2 = x._2._1
      val y2 = x._2._2
      val vectorResult = (x1 * x2 + y1 * y2) / ((Math.sqrt(x1 * x1 + y1 * y1)) * (Math.sqrt(x2 * x2 + y2 * y2)))
      if (vectorResult>1) {
        println(vectorResult+">>>>>>>>>>>>>>>>>")
      }
      val angle = Math.acos(vectorResult)
//      println(angle + "=========angel")

      (angle, threePoint)
    })
      .filter(_._1 > ((math.Pi) / 3))

    count = result.size
    (count, result.map(_._2))

  }

  /**
    * 加载案发数据
    *
    * @param spark
    * @param index
    * @param time
    * @param day
    * @param deviceCodes
    * @param codeType
    * @return
    */
  def loadIncidentData(spark: SparkSession, index: String, time: String, day: String, deviceCodes: java.util.ArrayList[String], codeType: String) = {
    val times = time.split("-")
    val starttime = day + " " + times(0)
    val endtime = day + " " + times(1)
    val esRDD = LoadESData.getMutliDeviceData(spark, index, deviceCodes, starttime, endtime)


    var incidentRdd: RDD[(String, String)] = spark.sparkContext.emptyRDD[(String, String)]
    if (esRDD.count() > 0) {
      incidentRdd = esRDD.map(e => {
        (e._2.get(codeType).get + "",
          codeType)
      }).distinct()
          .filter(x=>{
            x._1 !=None && !"None".equals(x._1)
          })
    }
    else {
      println("未找到相应的数据")
    }
    //    var incidentRdd: RDD[WifiEntity] = spark.sparkContext.emptyRDD[WifiEntity]
    //    if (esRDD.count() > 0) {
    //      incidentRdd = esRDD.map(e => {
    //        WifiEntity(
    //          e._1, e._2.get(codeType).get + "",
    //          e._2.get(codeType).get + "",
    //          codeType,
    //          Timestamp.valueOf(e._2.get("captime").get + ""),
    //          Timestamp.valueOf(e._2.get("captime").get + ""),
    //          //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
    //          new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
    //
    //          e._2.get("device_code").get + "",
    //          e._2.get("device_name").get + "",
    //          e._2.get("address").get + "",
    //          //            e._2.get("location")+"",
    //          e._2.get("captime").get + "",
    //          java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
    //          java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
    //          None,
    //          ""
    //        )
    //      })
    //
    //    }
    //    else {
    //      println("未找到相应的数据")
    //    }
    incidentRdd
  }

  /**
    * 加载徘徊分析数据
    *
    * @param spark
    * @param index
    * @param time
    * @param day
    * @param codeType
    */
  def loadLingerData(spark: SparkSession, index: String, time: String, day: String, codeType: String) = {
    val sdf_day = new SimpleDateFormat("yyyy-MM-dd")
    val times = time.split("-")
    val days = day.split(",")
    //获取连续天数
    //    val startday = sdf_day.parse(days(0))
    //    val endday = sdf_day.parse(days(1))
    //    val dayCount = (endday.getTime - startday.getTime) / (24 * 60 * 60 * 1000)

    var starttime = days(0) + " " + times(0)
    var endtime = days(1) + " " + times(1)
    val queryPara = new JSONObject()
    val queryBool = new JSONObject()
    val queryDevice = new JSONObject()
    val queryMatchAll = new JSONObject()
    queryMatchAll.put("match_all", queryDevice)
    val queryFilters = new JSONArray()
    queryFilters.add(queryMatchAll)
    if (starttime != null && !starttime.equals("") && endtime != null && !endtime.equals("")) {
      val queryTime = new JSONObject()
      queryTime.put("gte", starttime)
      queryTime.put("lte", endtime)
      val querycaptime = new JSONObject()
      querycaptime.put("captime", queryTime)
      val queryRange = new JSONObject()
      queryRange.put("range", querycaptime)
      queryFilters.add(queryRange)
    }
    queryBool.put("filter", queryFilters)
    queryPara.put("bool", queryBool)
    println(queryPara.toJSONString)
    val esData = EsSpark.esRDD(spark.sparkContext, index, queryPara.toJSONString).cache()
    var incidentRdd: RDD[WifiEntity] = spark.sparkContext.emptyRDD[WifiEntity]
    if (esData.count() > 0) {
      incidentRdd = esData
          .filter(x=>{
            x._2.get(codeType).get !=None
          })
          .filter(x=>{
            val captime = (x._2.get("captime").get+"").split(" ")(1)
            captime>times(0) && captime<times(1)
          })
        .map(e => {

        WifiEntity(
          e._1, e._2.get(codeType).get + "",
          e._2.get(codeType).get + "",
          codeType,
          Timestamp.valueOf(e._2.get("captime").get + ""),
          Timestamp.valueOf(e._2.get("captime").get + ""),
          //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
          new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),

          e._2.get("device_code").get + "",
          e._2.get("device_name").get + "",
          e._2.get("address").get + "",
          e._2.get("location").get + "",
          e._2.get("captime").get + "",
          java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
          java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
          e._2.get("phone").get + "",
          "",
          e._2.get("picpath").get+""
        )
      })


    }
    else {
      println("未找到相应的数据")
    }
    incidentRdd.filter(x=>{

      x.idNumber !=None && !"None".equals(x.idNumber)
    })
  }

  def updateStatus(jobId: Integer) = {
    val table = TABLE_LINGER_STATUS
    try {
      Class.forName("com.mysql.cj.jdbc.Driver")
    } catch {
      case e =>
        e.printStackTrace()
    }
    val connection = DriverManager.getConnection(JDBC_URL, USER_NAME, PASSWORD)
    val statement = connection.prepareStatement(
      s"update  $table set taskstatus = ?, successtime=? where id = ?  "
    )
    statement.setInt(1, 2)
    statement.setString(2, ParseTime.LongToDate(System.currentTimeMillis().toString))
//    statement.setInt(3, resultcount)
    statement.setInt(3, jobId)
    statement.executeUpdate()
    statement.close()
    connection.close()
  }
}

case class LingerCondition(
                            jobId: Integer,
//                            taskName: String,
                            codeType: String,
//                            incidentAddress: String,
                            incidentDay: String,
                            incidentTime: String,
                            hasCriminalRecord: Integer,
                            hasGang: Integer,
                            deviceCodes: String,
                            lingerDay: String,
                            lingerTime: String,
                            lingerCount: String
                          )

case class LingerResult(
                         resultid: String,
                         codetype: String,
                         count: Integer,
                         createtime: String,
                         jobid: Integer,
                         lingercount: Integer,
                         location: String,
                         phone: String,
                         tagcode: String
                       )

case class LingerDetail(
                         address: String,
                         captime: String,
                         codetype: String,
                         devicecode: Integer,
                         devicename: String,
                         latitude: Double,
                         location: String,
                         longtitude: Double,
                         phone: String,
                         resultid: String,
                         tagcode: String
                       )